package com.navinfo.tripanalysis.offline.service.impl;

import com.navinfo.tripanalysis.common.arithmetic.common.EngineAggregatedData;
import com.navinfo.tripanalysis.common.arithmetic.convert.EngineDataParquetConvert;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.offline.service.SaveTripEngineDataService;
import com.navinfo.tripanalysis.offline.service.TripEngineConvertService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;

/**
 * 保存行程发动机数据到hdfs中的hive指定分区中
 *
 * @author 刘俊
 */
public class SaveTripEngineDataServiceHiveImpl extends AbstractHiveSaveService implements SaveTripEngineDataService {
    static final Logger logger = LoggerFactory.getLogger(SaveTripEngineDataServiceHiveImpl.class);

    private TripEngineConvertService convertService;

    public SaveTripEngineDataServiceHiveImpl(TripEngineConvertService convertService) {
        this.convertService = convertService;
    }

    @Override
    public void save(SparkSession spark, JavaSparkContext jsc, JavaPairRDD<Long, List<EngineAggregatedData>> pairRDD, long d) {
        long start = System.currentTimeMillis();
        logger.error("进行行程发动机数据统计信息落盘，保存Hive开始...");

        String day = DateUtils.format(new Date(d), DateUtils.DateFormat.YYYYMMDD);
        super.saveData(spark, pairRDD, hdfsPath +"/day="+day, false, Collections.singletonList("day"),  Collections.singletonList("'"+day+"'"));
        logger.error("进行行程发动机数据统计信息落盘，保存Hive数据结束，耗{}ms", System.currentTimeMillis()-start);
    }

    @Override
    protected StructType createSchema() {
        return convertService.createSchema();
    }

    @Override
    protected JavaRDD<Row> createRowRDD(JavaPairRDD resourceRDD) {
        return resourceRDD.flatMap((FlatMapFunction<Tuple2<Long, List<EngineAggregatedData>>, Row>) longListTuple2 -> {
            List<EngineAggregatedData> srcList = longListTuple2._2;
            List<Row> list = new ArrayList<>(srcList.size());
            for (EngineAggregatedData m : srcList) {
                if (m != null) {
                    list.add(convertService.toRow(m, EngineDataParquetConvert.createSchema()));
                }
            }
            return list.iterator();
        });
    }
}
